package com.android.kotlincoroutinemvvmdemo
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.FlowCollector
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.flow.debounce
import kotlinx.coroutines.flow.drop
import kotlinx.coroutines.flow.dropWhile
import kotlinx.coroutines.flow.filterIsInstance
import kotlinx.coroutines.flow.flattenMerge
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.mapLatest
import kotlinx.coroutines.flow.mapNotNull
import kotlinx.coroutines.flow.onCompletion
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.onStart
import kotlinx.coroutines.flow.sample
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.flow.transform
import kotlinx.coroutines.flow.transformWhile
import kotlinx.coroutines.flow.zip
import kotlinx.coroutines.runBlocking
import org.junit.Test

class FlowDemo2 {
    @Test
    fun testTake() = runBlocking {
        //take  取前面n个元素
        flowOf(1, 2, 3, 4, 5, 6, 7)
            .onStart {
                doPrint("onStart")
            }
            .take(3)
            .onCompletion {
                //自然完成
                doPrint("onCompletion")
            }
            .catch {
                //抛出异常被迫完成
                doPrint("${it.message}")
            }
            .collect {
                doPrint(it)
        }
    }

    @Test
    fun testDrop() = runBlocking {
        //drop  丢弃前面n个元素
        flowOf("A","B","C","D","E").drop(2).collect {
            doPrint(it)
        }
    }

    @Test
    fun testMapLatest() = runBlocking {
        //mapLatest:当emit发送值时,会丢弃上一次还未转换完成的值.
        flow {
            repeat(10) {
                delay(10)
                emit(it)
            }
        }.mapLatest {
            delay(15)
            it
        }.collect {
            doPrint(it)
        }
    }

    @Test
    fun testMapNotNull() = runBlocking {
        //mapNotNull:仅发送map之后不为null的值
        flow {
            repeat(10) {
                emit(it)
            }
        }.mapNotNull {
            if (it > 5) {
                it
            } else {
                null
            }
        }.collect {
            doPrint(it)
        }
    }

    @Test
    fun testTransform() = runBlocking {
        //transform:把上游emit发送过来的值进行转换,之后再进行emit新值.
        flow {
            repeat(10) {
                delay(10)
                emit(it)
            }
        }.transform {
            delay(100)
            emit(it * 10)
        }.collect {
            doPrint(it)
        }
    }

    @Test
    fun testTransformWhile() = runBlocking {
        //transformWhile:当条件不满足时,后续的emit操作会终止.
        flow {
            repeat(10) {
                delay(10)
                emit(it)
            }
        }.transformWhile {
            emit(it)
            it != 2
        }.collect {
            doPrint(it)
        }
    }

    @Test
    fun testFilterInstance() = runBlocking {
        //filterIsInstance:筛选对应类型的值
        flow {
            emit(10)
            emit(null)
            emit("123")
        }.filterIsInstance<String>()
         .collect {
             doPrint(it)
         }
    }

    @Test
    fun testDropWhile() = runBlocking {
        //dropWhile:丢弃前面满足所有条件的数据,返回后面不满足条件的.
        flow {
            emit(1)
            emit(null)
            emit(false)
        }.dropWhile {
            it != null
        }.collect {
            doPrint(it)
        }
    }

    @Test
    fun testDebounce() = runBlocking {
        //debounce:指定时间内只接收最新的额值,其他的过滤掉.用于防抖 指定时间接收数据
        flow {
            emit(1)
            delay(100)

            emit(5)
            delay(50)

            emit(6)
            delay(300)

            emit(20)
            delay(120)

            emit(10)
            delay(80)
        }.debounce(300)
            .collect {
                doPrint(it)
            }
    }

    @Test
    fun testSample() = runBlocking {
        //sample:在指定周期内获取最新发出的值,定时周期接收.
        createFlow().collect(object : FlowCollector<Int> {
             override suspend fun emit(value: Int) {
                 doPrint(value)
             }
         })
    }

    @Test
    fun testZip() = runBlocking {
        //zip:压缩转换两个流
        val flow0 = flowOf(0, 2, 4).onEach { delay(50) }
        val flow1 = flowOf("a", "b", "c", "d").onEach { delay(150) }

        val startTime = System.currentTimeMillis()

        flow0.zip(flow1){ i, j -> "${i}${j}"}
             .collect {
                 doPrint("${it},耗时${System.currentTimeMillis() - startTime}秒")
             }
    }

    @Test
    fun testCombine() = runBlocking {
        //combine:合并转换两个流
        val flow1 = flow {
            emit("A")
            delay(150)
            emit("B")
        }

        val flow2 = flow {
            emit(1)
            delay(50)
            emit(2)
        }

        flow1.combine(flow2) { t1, t2 -> return@combine t1 to t2 }
             .collect {
                doPrint(it)
             }
    }

    @Test
    fun testFlattenMerge() = runBlocking {
        //flattenMerge:展平流,一个一个元素轮流emit.
        flow {
            emit(flowOf(1, 2, 3))
            emit(flowOf("A", "B", "C"))
        }.flattenMerge()
         .collect {
             doPrint(it)
         }
    }

    private fun createFlow(): Flow<Int> {
        return flow {
            emit(0)
            delay(100)

            emit(12)
            delay(200)

            emit(30)
            delay(20)

            emit(25)
            delay(80)
        }.sample(200)
    }

    companion object {
        fun doPrint(value: Any?) {
            println("result:${value}")
        }
    }
}